package com.niit.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.Scanner;

/**
 * Date:2024/11/7
 * Author：Ys
 * Description:
 */
public class HBaseClient {


    //定义配置对象
    private Configuration configuration;
    //定义连接对象
    private Connection connection;
    //定义操作者 拥有管理员的操作权限
    private Admin admin;

    //初始化操作
    @Before
    public void initClinet() throws IOException {
        //资源和连接初始化
        //利用HBaseConfiguration创建配置对象
        configuration = HBaseConfiguration.create();
        //利用已经创建好的配置对象进行连接
        connection = ConnectionFactory.createConnection(configuration);
        //创建管理员对象
        admin = connection.getAdmin();


    }

    //创建名称空间
    @Test  //@Before:是在@Test执行前会自动调用     @After：在@Test执行后会自动调用
    public void createNameSpace() {
        //1.调用方法创建名称空间
        NamespaceDescriptor.Builder builder = NamespaceDescriptor.create("2024BD1");

        //2.使用builder构造出来对应创建对象来完成名称空间的创建
        try{
            admin.createNamespace(builder.build());
        }catch (IOException e){
            System.out.println("名称空间已经存在");
            e.printStackTrace();
        }


    }
    //判断表是否存在
    @Test
    public void isTableExist() throws IOException {
        //1.声明一张表
        TableName tableName = TableName.valueOf("MOMO_CHAT","MSG");
        //2.操作者判断表是否存在
        boolean b = admin.tableExists(tableName);

        System.out.println("该表是否存在:"+b);

    }

    //创建表
    @Test
    public void createTable() throws IOException {
        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";
        //定义列族名
        String columnFamily = "C1";

        //2.判断表是否存在
        TableName tableName1 = TableName.valueOf(nameSpace,tableName);
        if(admin.tableExists(tableName1)){
            return;
        }

        //3.创建 表的描述构建器    ----想法  此时拥有创建表的一个意图
            //此项 我想在 2024BD1 下 创建  WATER_BILL 表
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName1);

        //创建  列族的描述构建器    ---想法 此时拥有创建列族的一个意图
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder( Bytes.toBytes(columnFamily) );

        //4. 将 创建列族的意图 付之于行动
        ColumnFamilyDescriptor columnZu = columnFamilyDescriptorBuilder.build();

        //5.在创建好的列族 放进 创建表的意图中
        tableDescriptorBuilder.setColumnFamily(columnZu);

        //6.将 创建表的意图 付之于行动
        TableDescriptor table = tableDescriptorBuilder.build();  //具有列族的真实表

        //7.利用操作者，将表创建在Hbase中
        admin.createTable(table);
    }

    @Test
    public void dropTable() throws IOException {
        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";

        //2.判断表是否存在
        TableName tableName1 = TableName.valueOf(nameSpace,tableName);
        // 判断表是否存在，如果存在就删除
        if(admin.tableExists(tableName1)){
            //先禁用
            admin.disableTable(tableName1);
            //再删除
            admin.deleteTable(tableName1);
        }

    }

    // 修改表， 修改列族C1的版本
    @Test
    public void modifyTable() throws IOException {

        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";
        //定义列族名
        String columnFamily = "C1";
        //版本
        int versions = 2;
        TableName tableName1 = TableName.valueOf(nameSpace,tableName);
        //2. 判断表是否存在，如果不存在就结束方法
        if(!admin.tableExists(tableName1)){
                return;
        }
        //3.得到要修改的真实表
        TableDescriptor descriptor = admin.getDescriptor(tableName1);

        //4.创建 真实表的 描述构建器  --- 在真实表上 先提出修改的意图
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(descriptor);

        //5. 得到要修改 哪个列族描述 真实的列族
        ColumnFamilyDescriptor columnFamily1 = descriptor.getColumnFamily(Bytes.toBytes(columnFamily));

        //6.1 通过 真实的列族 构建出 列族描述器
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(columnFamily1);
        //6.2 在 修改列族的想法中 进行修改
        columnFamilyDescriptorBuilder.setMaxVersions(versions);

        //7.修改好的列族 再转交给表的意图
        tableDescriptorBuilder.modifyColumnFamily(columnFamilyDescriptorBuilder.build());

        //8.修改表
        admin.modifyTable(tableDescriptorBuilder.build());
    }

    //添加数据 put '表名','rowkey','列族:列名','列值'
    @Test
    public void addData() throws IOException {
        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";
        //定义列族名
        String columnFamily = "C1";
        //定义RowKey
        String rowKey = "1001";
        //定义列名
        String colName = "NAME";
        //列值
        String colValue = "狗剩";

        //2.构建 put 对象 --》 put 命令
        Put put = new Put( Bytes.toBytes(rowKey) );

        //3.添加 列族 列名 和列值
        put.addColumn(  Bytes.toBytes(columnFamily),     Bytes.toBytes(colName),         Bytes.toBytes(colValue)      );

        //4.使用HBase连接获取Table
        TableName tableName1 = TableName.valueOf(nameSpace, tableName);
        Table table = connection.getTable(tableName1);

        table.put(put);
        table.close();

    }
    //根据RowKey 查询数据
    @Test
    public void getDataByRowKey() throws IOException {
        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";

        //定义RowKey
        String rowKey = "1001";

        //1.通过TableName 获取表
        TableName tableName1 = TableName.valueOf(nameSpace, tableName);
        //2.连接表  返回的是一张可操作的表
        Table table = connection.getTable(tableName1);
        //3.构建一个查询对象
        Get get = new Get( Bytes.toBytes(rowKey) );
        //4.利用已经创建好的 可操作的表 去进行查询
        Result result = table.get(get);//result 是一个结果集 里面会有很多数据
        //5.通过结果集获得当前的Rowkey
        System.out.println( "rowkey===>" + Bytes.toString(result.getRow()) );
        //6.通过结果集获得所有单元格
        List<Cell> cells = result.listCells();
        //7.遍历所有单元格 获得每个单元格中的信息
        for (Cell cell:cells) {

            //打印列族名                              列族数组                      该数据的起始值            该数据的结束值
            System.out.println("列族：" + Bytes.toString(  cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength() )    );

            //打印列名                              列数组                      该数据的起始值            该数据的结束值
            System.out.println("列名：" + Bytes.toString(  cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength() )    );

            //打印列值
            System.out.println( "列值==>" + Bytes.toString(  cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()     ) );
        }
        //8.关闭表
        table.close();


    }

    //根据 Rowkey 删除数据  delete '表名','rowkey','列族：列'
    @Test
    public void deleteDataByRowKey() throws IOException {
        //1.定义名称空间
        String nameSpace = "2024BD1";
        //定义表名
        String tableName = "WATER_BILL";

        //定义RowKey
        String rowKey = "1001";

        //1.通过TableName 获取表
        TableName tableName1 = TableName.valueOf(nameSpace, tableName);
        //2.连接表  返回的是一张可操作的表
        Table table = connection.getTable(tableName1);

        //3. 创建Delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        //4.添加列族：列
        delete.addColumn(  Bytes.toBytes("C1"),Bytes.toBytes("NAME") );

        //5.执行delete操作
        table.delete(delete);

        //6.关闭表连接
        table.close();
    }
    //扫描 MOMO_CHAT：MSG 表
    @Test
    public void scanData() throws IOException {

        String nameSpace = "MOMO_CHAT";
        String tableName = "MSG";

        TableName tableName1 = TableName.valueOf(nameSpace, tableName);
        Table table = connection.getTable(tableName1);

        //1.构建scan对象
        Scan scan = new Scan();
        scan.setLimit(50);

        //2.可操作表 传入 scan对象来获得Scanner
        ResultScanner scanner = table.getScanner(scan);
        //ResultScanner 记录了多行数据   Result单行数据
        for (Result result: scanner){
            //List<Cell> cells = = result.listCells();
             Cell[] cells = result.rawCells();
            for (Cell cell:cells){
                //打印列族名                              列族数组                      该数据的起始值            该数据的结束值
                System.out.println("列族：" + Bytes.toString(  cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength() )    );

                //打印列名                              列数组                      该数据的起始值            该数据的结束值
                System.out.println("列名：" + Bytes.toString(  cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength() )    );

                //打印列值
                System.out.println( "列值==>" + Bytes.toString(  cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()     ) );

            }
        }

        table.close();

    }


    //释放资源操作 关闭连接
    @After
    public void closeClinet() throws IOException {
        //关闭连接

        admin.close();
        connection.close();
    }
}
